package com.ctrip.ibu.member.client.common;


import com.alibaba.fastjson.JSONObject;
import com.ctrip.ibu.member.client.module.AllBrokerBriefInfo;
import com.ctrip.ibu.member.client.module.MessageInfo;
import com.ctrip.ibu.member.client.module.RequestQueueInfo;
import com.ctrip.ibu.member.client.module.TopicMessage;
import com.ctrip.ibu.member.common.result.HandleResult;
import com.ctrip.ibu.member.common.util.HttpClientUtil;
import com.google.common.collect.Lists;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;
import java.util.Objects;
import java.util.UUID;

/**
 * @author he.shaorong
 * @create 2023/3/27 16:31
 * @since 1.0.0
 */
public class BrokerStatusUtil {
    private static final Logger logger = LoggerFactory.getLogger(BrokerStatusUtil.class);

    static final List<String> nameServerIpList = Lists.newArrayList("http://10.32.114.131:8081");
    static final String GET_ALIVE_BROKER_LIST_URL = ":8081/getAliveBrokerList";

    static final String PUT_QUEUE = ":8082/putQueue";
    static final String GET_QUEUE = ":8082/getQueue";

    static final String GET_QUEUE_CALL_BACK = ":8082/getQueueCallback";

    /**
     * 查询可用的broker列表
     *
     * @return
     */
    public static AllBrokerBriefInfo getAliveBrokerIpList() {
        AllBrokerBriefInfo allBrokerBriefInfo = null;
        try {
            JSONObject aliveBrokerListJson = HttpClientUtil.get(nameServerIpList.get(0) + GET_ALIVE_BROKER_LIST_URL);
            if (Objects.nonNull(aliveBrokerListJson)) {
                allBrokerBriefInfo = JSONObject.parseObject(String.valueOf(aliveBrokerListJson), AllBrokerBriefInfo.class);
            }
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
        return allBrokerBriefInfo;
    }

    /**
     * 发送生产消息给broker
     *
     * @param ip
     * @param topicMessage
     * @return 发送结果
     */
    public static boolean sendToMessageBroker(String ip, TopicMessage topicMessage) {
        String sendUrl = "http://" + ip + PUT_QUEUE;
        try {
            HttpClientUtil.postJson(sendUrl, JSONObject.toJSONString(topicMessage));
            //正确响应则代表消息正常发出
        } catch (Exception e) {
            logger.error("sendToMessageBroker error! e:", e);
            throw new RuntimeException(e);
        }
        return true;

    }

    /**
     * 从broker消费消息
     *
     * @param ip
     * @param topic
     * @return 消息内容
     */
    public static MessageInfo consumeMessageFromBroker(String ip, String topic) {
        MessageInfo messageInfo;
        String sendUrl = "http://"+ ip + GET_QUEUE;
        try {
            RequestQueueInfo requestQueueInfo = new RequestQueueInfo();
            requestQueueInfo.setTopic(topic);
            requestQueueInfo.setGroupId("xx");
            requestQueueInfo.setSerialNumber(UUID.randomUUID().toString());
            JSONObject aliveBrokerListJson = HttpClientUtil.postJson(sendUrl, JSONObject.toJSONString(requestQueueInfo));
            messageInfo = JSONObject.parseObject(String.valueOf(aliveBrokerListJson), MessageInfo.class);
            if(Objects.nonNull(messageInfo) && !messageInfo.isAllFinished() && messageInfo.isSupportThisTopic()) {
                //
                String callbackUel = "http://"+ ip + GET_QUEUE_CALL_BACK;

                JSONObject callback = HttpClientUtil.postJson(callbackUel, JSONObject.toJSONString(requestQueueInfo));
                HandleResult handleResult = JSONObject.parseObject(String.valueOf(callback), HandleResult.class);
                if(!handleResult.isHandleResult()) {
                    logger.error("get Queue Callback error!");
                }
            }


        } catch (Exception e) {
            logger.error("consumeMessageFromBroker error!", e);
            throw new RuntimeException(e);
        }
        return messageInfo;

    }
}
